package com.yanggu.kafka.timer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class CronTimer {

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    //配置监听器, 消费third这个topic中的数据
    //autoStartup = "false", 禁止kafkaListener自启动
    @KafkaListener(id = "timingConsumer", topics = {"third"}, autoStartup = "false")
    public void onMessage(ConsumerRecord<?, ?> record) {
        System.out.println("消费成功" + record);
    }

    //定时启动器
    @Scheduled(cron = "0 49 13 * * ?")
    public void startListener() {
        System.out.println("启动监听器");
        if (!registry.getListenerContainer("timingConsumer").isRunning()) {
            registry.getListenerContainer("timingConsumer").start();
            System.out.println("监听器启动成功");
        }
    }

    //定时关闭监听器
    @Scheduled(cron = "0 51 13 * * ?")
    public void stopListener() {
        System.out.println("关闭监听器");
        registry.getListenerContainer("timingConsumer").resume();
    }

}
